package com.xavier.dong.nacos.grpc.client.grpc;

import com.google.common.collect.Lists;
import com.xavier.dong.nacos.grpc.api.*;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Component;

import java.security.SecureRandom;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Example class demonstrating the usage of {@link GrpcClient}s inside an application.
 */
@Component
@Slf4j
public class GrpcClientService {

    /**
     * 阻塞 服务端流(同步)
     */
    @GrpcClient("spring-boot-nacos-grpc-server")
    private HelloStreamGrpc.HelloStreamBlockingStub helloStreamBlockingStub;

    /**
     * 非阻塞 服务端流(异步)
     */
    @GrpcClient("spring-boot-nacos-grpc-server")
    private HelloStreamGrpc.HelloStreamStub helloStreamStub;

    /**
     * 发送简单rpc (同步)
     *
     * @param name 名字
     * @return {@link String}
     */
    public String sendSimpleRpc(final String name) {
        try {
            final SimpleFeature simpleFeature = this.helloStreamBlockingStub.simpleRpc(Simple.newBuilder().setName(name).build());
            return simpleFeature.getName();
        } catch (final StatusRuntimeException e) {
            log.error("Request failed", e);
            return "FAILED with " + e.getStatus().getCode();
        }
    }

    /**
     * 发送请求,响应stream(同步)
     *
     * @param name 名字
     * @return {@link Iterator}<{@link SimpleFeature}>
     */
    public List<String> sendServer2ClientRpc(final List<String> name) {
        try {
            SimpleList simpleList = SimpleList.newBuilder()
                    .addAllSimpleList(name.stream().map(e -> Simple.newBuilder().setName(e).build()).collect(Collectors.toList()))
                    .build();
            final Iterator<SimpleFeature> featureIterator = this.helloStreamBlockingStub.server2ClientRpc(simpleList);
            List<String> list = Lists.newArrayList();
            while (featureIterator.hasNext()) {
                list.add(featureIterator.next().getName());
            }
            return list;
        } catch (final StatusRuntimeException e) {
            log.error("Request failed", e);
            return Collections.emptyList();
        }
    }


    /**
     * 发送stream,响应实体(异步)
     *
     * @param count 数
     */
    public void sendClient2ServerRpc(int count) throws InterruptedException {

        SecureRandom random = new SecureRandom();
        log.info("request client2ServerRpc {}", count);
        final CountDownLatch finishLatch = new CountDownLatch(1);

        final StreamObserver<Simple> requestObserver = this.helloStreamStub.client2ServerRpc(new StreamObserver<SimpleSummary>() {

            @Override
            public void onNext(SimpleSummary simpleSummary) {
                // 返回SimpleSummary
                log.info("client2ServerRpc onNext : {}", simpleSummary.getFeatureCount());
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("client2ServerRpc error : {}", Status.fromThrowable(throwable));
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.error("client2ServerRpc finish");
                finishLatch.countDown();
            }
        });

        try {
            for (int i = 0; i < count; i++) {
                log.info("simple : {}", i);
                Simple simple = Simple.newBuilder().setName("client2ServerRpc" + i).setNum(i).build();
                requestObserver.onNext(simple);
                Thread.sleep(random.nextInt(200) + 50);
            }
        } catch (RuntimeException e) {
            // Cancel RPC
            requestObserver.onError(e);
            throw e;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        // 结束请求
        requestObserver.onCompleted();

        // Receiving happens asynchronously
        if (!finishLatch.await(1, TimeUnit.MINUTES)) {
            log.error("client2ServerRpc can not finish within 1 minutes");
        }

    }


    /**
     * 发送stream,响应实体(异步)
     *
     * @param count 数
     */
    public void sendBindirectionalStreamRpc(int count) throws InterruptedException {

        log.info("request bindirectionalStreamRpc");
        final CountDownLatch finishLatch = new CountDownLatch(1);

        final StreamObserver<Simple> requestObserver = this.helloStreamStub.bindirectionalStreamRpc(new StreamObserver<Simple>() {
            @Override
            public void onNext(Simple simple) {
                log.info("bindirectionalStreamRpc receive message : {}", simple.getName());
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("bindirectionalStreamRpc Failed: {0}", Status.fromThrowable(throwable));
                finishLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("Finished bindirectionalStreamRpc");
                finishLatch.countDown();
            }
        });

        try {
            Simple[] requests = {newSimple("1"), newSimple("2"), newSimple("3"), newSimple("4")};

            for (Simple request : requests) {
                log.info("Sending message {}", request);
                requestObserver.onNext(request);
            }
        } catch (RuntimeException e) {
            // Cancel RPC
            requestObserver.onError(e);
            throw e;
        }
        requestObserver.onCompleted();

        if (!finishLatch.await(1, TimeUnit.MINUTES)) {
            log.error("routeChat can not finish within 1 minutes");
        }
    }

    private Simple newSimple(String name) {
        return Simple.newBuilder().setName(name).build();
    }

}